package de.dopichaj.labrador.index.backend;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;

import org.apache.log4j.Logger;

import de.dopichaj.labrador.util.io.BufferedDataInputFromRandomAccessFile;


public final class DiskInvertedList implements InvertedList {

    private static final Logger log = Logger.getLogger(DiskInvertedList.class);
    
    private final RandomAccessFile file;
    private final RandomAccessFile tempFile;
    private final TermFrequencyOutputStream tempOutput;
    private final File tempFileName;
    private final File tempFileName2;
    private final int runLength;
    private final PriorityQueue<TermFrequency> buffer;
    /**
     * Flag whether to store the term IDs in the final run; this is mostly
     * useful for testing and debugging.
     */
    private final boolean storeTermIDs;
    
    /**
     * The start positions of the sorted runs in the temporary file.
     */
    private List<Long> bufferPositions = new ArrayList<Long>();
    private TermFrequency lastWritten = null;
    private int realRunLength = 0;
    
    
    public DiskInvertedList(final File tempDir, final RandomAccessFile file,
        final int runLength, final boolean storeTermIDs) throws IOException {
        
        assert file != null;
        assert runLength > 0;
        
        this.file = file;
        this.runLength = runLength;
        buffer = new PriorityQueue<TermFrequency>(runLength);
        
        tempFileName = File.createTempFile("dtil", null, tempDir);
        tempFileName.deleteOnExit();
        tempFileName2 = File.createTempFile("dtil", null, tempDir);
        tempFileName2.deleteOnExit();

        this.tempFile = new RandomAccessFile(tempFileName, "rw");
        this.tempOutput = new TermFrequencyOutputStream(tempFile);
        this.storeTermIDs = storeTermIDs;
    }

    public void addTermFrequency(TermFrequency freq) throws IOException {
        
        if (freq.getFrequency() == 0) {
            log.warn("Adding term with 0 frequency: " + freq);
        }
        addToBuffer(freq);
    }

    private void addToBuffer(final TermFrequency freq) throws IOException {
        
        // if we have not started writing a run, add the new item to the buffer
        if (lastWritten == null) {
            
            buffer.add(freq);
            
        } else {
            
            // if the new item is greater than the last written run element
            if (freq.compareTo(lastWritten) > 0) {
                
                // put it in the buffer
                buffer.add(freq);
                
            } else {
                
                // flush the remaining buffer
                realRunLength += buffer.size();
                log.info("Wrote run of length " + realRunLength);
                while (!buffer.isEmpty()) {
                    tempOutput.writeTermFrequency(buffer.poll());
                }

                // add the new item to the buffer
                buffer.add(freq);
                realRunLength = 1;
                
                // remember that there are no unwritten items in this run
                lastWritten = null;

                tempOutput.terminateRun();
            }
        }
        
        // if the buffer is full
        if (buffer.size() == runLength) {
            
            // remember the start of the run in the run file
            if (lastWritten == null) {
                tempOutput.flush();
                bufferPositions.add(tempFile.getFilePointer());
            }
            
            // flush the first item and remember it
            lastWritten = buffer.poll();
            tempOutput.writeTermFrequency(lastWritten);
            realRunLength++;
        }
    }

    private void flush() throws IOException {
        
        if (buffer.size() > 0) {
            
            // if there is no open run
            if (lastWritten == null) {
                
                // remember the start position
                tempOutput.flush();
                bufferPositions.add(tempFile.getFilePointer());
                
            }

            // write the run
            realRunLength += buffer.size();
            log.info("Wrote run of length " + realRunLength);
            while (!buffer.isEmpty()) {
                tempOutput.writeTermFrequency(buffer.poll());
            }
            realRunLength = 0;
            
            lastWritten = null;
        }
        tempOutput.terminateRun();
        tempOutput.flush();
    }

    /**
     * Sort and finalize this inverted list.
     * 
     * @param dictionary
     * @throws IOException
     */
    public void sort(final Dictionary dictionary) throws IOException {
        
        flush();
        log.info("sorting");
        
        List<Long> currentBufferPositions = bufferPositions;
        currentBufferPositions.add(tempFile.getFilePointer());
        
        // create new temporary file
        RandomAccessFile output = tempFile;
        RandomAccessFile input = new RandomAccessFile(tempFileName2, "rw");
        
        // while there is more than one run
        while (currentBufferPositions.size() > 2) {

            log.info("Merging: " + currentBufferPositions.size() + " remaining");

            List<Long> newBufferPositions =
                new ArrayList<Long>((currentBufferPositions.size() + 1) / 2);
            
            // swap input and output files
            final RandomAccessFile oldInput = input;
            input = output;
            output = oldInput;
            output.seek(0);
            output.setLength(0);
            
            // make sure the runs are roughly the same size
            final int runCount = currentBufferPositions.size() - 1;
            final int newRunCount = (runCount + runLength - 1) / runLength;

            // merge runs
            for (int i = 0; i < newRunCount; i++) {

                // create streams for the runs to be merged
                final List<TermFrequencyInputStream> streams =
                    new ArrayList<TermFrequencyInputStream>(runLength);
                for (int j = 0; i + j < runCount; j += newRunCount) {
                    final long start = currentBufferPositions.get(j + i);
                    final long end = currentBufferPositions.get(j + i + 1);
                    streams.add(new TermFrequencyInputStream(
                        new BufferedDataInputFromRandomAccessFile(input,
                        start, end)));
                }
                
                // merge them
                newBufferPositions.add(output.getFilePointer());
                merge(streams, new TermFrequencyOutputStream(output));
            }
            newBufferPositions.add(output.getFilePointer());
            
            currentBufferPositions = newBufferPositions;
        }
        
        // copy last output file to permanent de.dopichaj.labrador.index file
        output.seek(0);
        final BufferedDataInputFromRandomAccessFile bufferedInput =
            new BufferedDataInputFromRandomAccessFile(output, 0, Long.MAX_VALUE);
        final TermFrequencyInputStream varInput = 
            new TermFrequencyInputStream(bufferedInput);
        final TermFrequencyOutputStream finalOutput =
            new TermFrequencyOutputStream(file, true, storeTermIDs);
        file.seek(0);
        log.info("About to copy " + output.length() + " bytes ...");
        int previousTermID = -1;
        TermFrequency freq;
        while ((freq = varInput.readTermFrequency()) != null) {
            
            // if this is the start of a new run
            if (dictionary != null && freq.getTermID() != previousTermID) {
                
                // if this is not the first run, terminate it
                if (previousTermID != -1) {
                    finalOutput.terminateRun();
                    finalOutput.flush();
                }
                
                // remember the file position for this term
                final long position = file.getFilePointer();
                assert position <= Integer.MAX_VALUE;
                dictionary.setIndexPosition(freq.getTermID(), (int)position);
                previousTermID = freq.getTermID();
            }
            finalOutput.writeTermFrequency(freq);
        }
        finalOutput.terminateRun();
        finalOutput.flush();
        file.seek(0);
        
        // clean up
        tempFile.close();
        tempFileName.delete();
    }
    
    public void sort() throws IOException {
        sort(null);
    }

    private void merge(final List<TermFrequencyInputStream> streams,
        final TermFrequencyOutputStream output) throws IOException {

        assert output != null;

        if (log.isDebugEnabled()) {
            log.debug("Merging " + streams.size() + " runs");
        }
        
        // fill heap
        assert buffer.isEmpty();
        final Map<TermFrequency, TermFrequencyInputStream> map =
            new HashMap<TermFrequency, TermFrequencyInputStream>();
        for (final TermFrequencyInputStream stream : streams) {
            final TermFrequency freq = stream.readTermFrequency();
            assert freq.getFrequency() > 0 : "Frequency <= 0: " + freq;
            if (freq != null) {
                map.put(freq, stream);
                buffer.add(freq);
            }
        }
        assert buffer.size() == streams.size();
        
        // while there are non-exhausted runs
        while (!streams.isEmpty()) {
            
            // output the smallest item
            final TermFrequency freq = buffer.poll();
            output.writeTermFrequency(freq);
            
            // re-fill the position in the heap
            final TermFrequencyInputStream stream = map.get(freq);
            assert stream != null : "Unknown stream for " + freq;
            map.remove(freq);
            assert !map.containsValue(stream);
            final TermFrequency newFreq = stream.readTermFrequency();
            if (newFreq != null) {
                assert newFreq != freq;
                map.put(newFreq, stream);
                buffer.add(newFreq);
            } else {
                streams.remove(stream);
            }
            
            assert map.size() == streams.size() : "map.size()=" + map.size() +
                ", streams.size()=" + streams.size() + ", buffer.size()=" + buffer.size();
        }
        
        output.terminateRun();
        output.flush();
    }

    private final class TFIterator implements Iterator<TermFrequency> {

        private final TermFrequencyInputStream input;
        private TermFrequency next;
        
        public TFIterator(final TermFrequencyInputStream input) throws IOException {
            this.input = input;
            this.next = input.readTermFrequency();
        }
        
        public boolean hasNext() {
            return next != null;
        }

        public TermFrequency next() {
            
            try {
                final TermFrequency result = next;
                next = input.readTermFrequency();
                return result;
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        public void remove() {
            throw new UnsupportedOperationException();
        }
    }
    
    public Iterator<TermFrequency> iterator() {
        if (!storeTermIDs) {
            throw new IllegalStateException("Cannot iterate over inverted " +
                "lists without term IDs.");
        }
        try {
            file.seek(0);
            return new TFIterator(new TermFrequencyInputStream(file, true));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void close() throws IOException {
        file.close();
    }
}
/*
Copyright (c) 2007 Philipp Dopichaj

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
*/